package mapred;

import java.io.IOException;

import mapred.config.CombineConfig;
import mapred.config.Compress;
import mapred.config.InputConfig;
import mapred.config.MapConfig;
import mapred.config.OutputConfig;
import mapred.config.ReduceConfig;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.KeyValueTextInputFormat;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.Reducer;
import org.apache.hadoop.mapred.TextOutputFormat;

@SuppressWarnings("unchecked")
public abstract class SimpleJob extends Job {

    public SimpleJob(String jobName, String... settings) throws IOException {
        super(jobName, settings);
    }

    @Override
    protected CombineConfig getCombineConfig() {
        return CombineConfig.newInstance(getCombinerClass());
    }
    
    protected abstract Class<? extends Reducer> getCombinerClass();

    @Override
    protected InputConfig getInputConfig(FileSystem fs) throws IOException {
        return InputConfig.newInstance(KeyValueTextInputFormat.class, this
                .getInputPaths(fs));
    }

    protected abstract String[] getInputPaths(FileSystem fs) throws IOException;

    @Override
    protected MapConfig getMapConfig() {
        return MapConfig.newInstance(this.getMapperClass(), Text.class,
                Text.class);
    }

    protected abstract Class<? extends Mapper> getMapperClass();

    @Override
    protected OutputConfig getOutputConfig(FileSystem fs) throws IOException {
        return OutputConfig.newInstance(TextOutputFormat.class, Text.class,
                Text.class, this.getOutputPath(fs), this.getCompress());
    }

    protected abstract String getOutputPath(FileSystem fs) throws IOException;

    protected abstract Compress getCompress();

    @Override
    protected ReduceConfig getReduceConfig() {
        return ReduceConfig.newInstance(this.getReducerClass(), this.getReducerNumber());
    }

    protected abstract Class<? extends Reducer> getReducerClass();

    protected abstract int getReducerNumber();
    
}
